package net.bwie.realtime.vehicle2.function;

import net.bwie.realtime.vehicle.bean.VehicleData;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;

/**
 * @Author: FuHe
 * @Date: 2025/6/8
 */
public class VehicleProvinceFunction extends ProcessWindowFunction<VehicleData, String, String, TimeWindow> {
    @Override
    public void process(String s, ProcessWindowFunction<VehicleData, String, String, TimeWindow>.Context context, Iterable<VehicleData> elements, Collector<String> out) throws Exception {
        HashSet<String> strings = new HashSet<>();
        for (VehicleData element : elements) {
            if (element.getVin() != null && !strings.contains(element.getVin())) {
                strings.add(element.getVin());
            }
        }
        out.collect(s + "," + strings.size());
    }
}
